package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * Created by Administrator on 2017/4/25 0025.
  * 最短路径距离算法（不包含最短路径）
  */
object TestDijstra {
  //针对有向图的Dijstra算法实现。
  def dijkstra[VD](g: Graph[VD, Double], origin: VertexId): Graph[(VD, Double),Double] = {
    //1,初始化
    var g2 = g.mapVertices{
      case(vid, _) =>
        val vd = if(vid == origin) 0 else Double.MaxValue
        (false, vd)
    }

    //2，遍历所有的点
    (0L until g.vertices.count).foreach{ i: Long =>

      //3,确定最短路径值最小的作为当前顶点。当然了.filter(!_._2._1)操作，确保了筛选范围是没有访问过的顶点。
      //filter()中的_表示任意一个顶点，其结构是（顶点ID，顶点属性）；_._2表示任意一个顶点的顶点属性；_._2._1表示任意一个顶点的顶点属性的第一个值，是布尔类型。
      val currentVertexId: VertexId = g2.vertices.filter(!_._2._1)
              .fold((0L, (false,Double.MaxValue))){
                case (a,b) => if(a._2._2 < b._2._2) a else b
              }._1
      //上面fold(){}返回的是一个顶点。

      //4，向与当前顶点相邻的顶点发消息，再聚合消息：取小值做为最短路径值。
      val newDistances: VertexRDD[Double] = g2.aggregateMessages[Double](
        //sendMsg:向临边发消息，内容为边的距离与最短路径之和。
        ctx => if(ctx.srcId == currentVertexId) ctx.sendToDst(ctx.srcAttr._2 +ctx.attr),
        (a,b) => math.min(a,b)
      )

      //5，生成结果图
      //vd表示旧的(g2)的顶点的属性，newSum表示新的(newDistances)的属性。
      g2 = g2.outerJoinVertices(newDistances){ (vid, vd, newSum) =>
        (vd._1 || vid == currentVertexId, math.min(vd._2, newSum.getOrElse(Double.MaxValue)))
      }
    }
    g.outerJoinVertices(g2.vertices){(vid, vd, dist) =>
      (vd, dist.getOrElse((false,Double.MaxValue))._2)
    }
  }

  //main方法
  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    val myVertices = sc.makeRDD(Array((1L,"A"),(2L,"B"),(3L,"C"),(4L,"D"),(5L,"E"),(6L,"F"),(7L,"G")))
    val myEdges = sc.makeRDD(Array(Edge(1L,2L,7.0),Edge(1L,4L,5.0),Edge(2L,3L,8.0),Edge(2L,4L,9.0),
      Edge(2L,5L,7.0),Edge(3L,5L,5.0),Edge(4L,5L,15.0),Edge(4L,6L,6.0),
      Edge(5L,6L,8.0),Edge(5L,7L,9.0),Edge(6L,7L,11.0)))
    val myGraph = Graph(myVertices,myEdges)
    val list_t = dijkstra(myGraph, 1L).vertices.collect
    for(t <- list_t){
      println(t)
    }
  }
}
